package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;

import static com.hmdp.utils.RedisConstants.SECKILL_CONSUMER;
import static com.hmdp.utils.RedisConstants.SECKILL_GROUP;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    private static final String ICR_ORDER_ID_PREFIX = "order";

    @Autowired
    private ISeckillVoucherService seckillVoucherService;
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Autowired
    private IVoucherOrderService voucherOrderService;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedissonClient redissonClient;

    /**
     * 线程池，处理秒杀任务
     */
    @Resource(name = "voucherExecutor")
    private ThreadPoolExecutor SECKILL_ORDER_EXECUTOR;

    /**
     * 判断秒杀资格的 LUA 脚本
     */
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    /**
     * 阻塞队列，默认大小为 Integer 最大值
     */
    private static final BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue(1024 * 1024);
    /**
     * 名称
     */
    private static final String QUEUE_NAME = "stream.orders";

    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    @PostConstruct
    public void init() {
        // 1、redis创建队列，创建消费者组，查看是否有消费者组
        // 创建redis订单队列
        RecordId recordId = stringRedisTemplate.opsForStream().add(QUEUE_NAME, Collections.singletonMap("k1","v1"));
        stringRedisTemplate.opsForStream().delete(QUEUE_NAME, recordId);
        Iterator<StreamInfo.XInfoGroup> iterator = stringRedisTemplate.opsForStream()
                .groups(QUEUE_NAME).stream().iterator();
        Boolean isFlag = false;
        while (iterator.hasNext()) {
            if (iterator.next().groupName().equals(SECKILL_GROUP)) {
                isFlag = true;
                break;
            }
        }
        if (!isFlag) {
            stringRedisTemplate.opsForStream().createGroup(QUEUE_NAME, SECKILL_GROUP);
        }

        // submit 可以执行 Callable 或者 runnable，返回 future，可通过 future 返回异常
        // excute 只能执行 runnable，直接返回异常信息
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    private class VoucherOrderHandler implements Runnable {

        @Override
        public void run() {
            while (true) {
                try {
                    // 1、获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
                    // <String,objects,object> => <streamKey,Map<k,v>>
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from(SECKILL_GROUP, SECKILL_CONSUMER),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(QUEUE_NAME, ReadOffset.lastConsumed())
                    );
                    // 2、判断消息是否获取成功
                    // 2.1、如果获取失败，说明没消息，继续下一次循环
                    if (list == null || list.isEmpty()) {
                        continue;
                    }
                    // 3、获取成功，进行下单
                    MapRecord<String, Object, Object> record = list.get(0);
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(record.getValue(), new VoucherOrder(), true);
                    handleVoucherOrder(voucherOrder);
                    // 4、ACK 确认
                    stringRedisTemplate.opsForStream().acknowledge(QUEUE_NAME, SECKILL_GROUP, record.getId());
                } catch (Exception e) {
                    log.error("处理订单异常", e);
                    handlePendingList();
                }
            }
        }

        private void handlePendingList() {
            while (true) {
                try {
                    // 1、获取 pending-list 中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream.orders 0
                    // <String,objects,object> => <streamKey,Map<k,v>>
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from(SECKILL_GROUP, SECKILL_CONSUMER),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create(QUEUE_NAME, ReadOffset.from("0"))
                    );
                    // 2、判断消息是否获取成功
                    if (list == null || list.isEmpty()) {
                        // 2.1、如果获取失败，说明 pending-list 没异常消息，结束循环
                        break;
                    }
                    // 3、获取成功，进行下单
                    MapRecord<String, Object, Object> record = list.get(0);
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(record.getValue(), new VoucherOrder(), true);
                    handleVoucherOrder(voucherOrder);
                    // 4、ACK 确认
                    stringRedisTemplate.opsForStream().acknowledge(QUEUE_NAME, SECKILL_GROUP, record.getId());
                } catch (Exception e) {
                    log.error("处理pending-list订单异常", e);
                    try {
                        Thread.sleep(20);
                    } catch (InterruptedException interruptedException) {
                        interruptedException.printStackTrace();
                    }
                }
            }
        }
    }

    /**
     * @param voucherOrder
     */
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        // 1、获取 userId
        Long userId = voucherOrder.getUserId();
        // 2、创建锁对象
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        // 3、获取锁
        // 参数依次为 waitTime（等待获取锁的时间，可在该时间内重复获取锁）,leaseTime（锁的时间）,TimeUnit（单位）
        // 不写参数则默认只获取一次锁，30s
        boolean isLock = lock.tryLock();
        // 3.1、判断是否拿到锁
        if (!isLock) {
            log.error("不允许重复下单：{}", userId);
            return;
        }
        // 3.2、处理订单信息
        try {
            // springboot 解决了循环调用
            voucherOrderService.createVoucherOrder(voucherOrder);
        } finally {
            // 即便 try 的代码有 return，finally 还是会执行
            lock.unlock();
        }
    }

    /**
     * 对数据库，根据订单信息，保存订单数据，修改库存。
     *
     * @param voucherOrder
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        // 4、一人一单
        Long userId = voucherOrder.getUserId();
        Long voucherId = voucherOrder.getVoucherId();
        Integer count = this.query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        if (count > 0) {
            log.error("用户已经购买过一次了,{}", userId);
            return;
        }
        // 5、扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1")
                .eq("voucher_id", voucherId)
                .gt("stock", 0)
                .update();
        if (!success) {
            log.error("库存不足，{}", voucherId + "_" + "transactional");
            return;
        }
        // 6、保存订单
        this.save(voucherOrder);
    }

    /**
     * 基于redis消息队列进行的下单
     *
     * @param voucherId
     */
    @Override
    public Result seckillVoucher(Long voucherId) {
        // 1、执行 LUA，判断秒杀资格，将订单信息放到消息队列里
        // 1.1、获取 userId 和 orderId
        Long userId = UserHolder.getUser().getId();
        Long orderId = redisIdWorker.nextId("order");
        // 1.2、xadd key *|id userId userId voucherId voucherId id orderId
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT
                , Collections.emptyList()
                , voucherId.toString()
                , userId.toString()
                , orderId.toString());
        // 2、判断结果不为 0，没有秒杀资格
        int r = result.intValue();
        if (r != 0) {
            return Result.fail(r == 1 ? "没有库存" : "不能重复下单");
        }
        return Result.ok(orderId);
    }

    /**
     * LUA 辅助 redis 判断秒杀资格，设置订单信息，放进队列给异步线程处理。
     *
     * @param voucherId
     */
    /*@Override
    public void seckillVoucher(Long voucherId) {
        // 1、执行 LUA，判断秒杀资格
        Long userId = UserHolder.getUser().getId();
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT
                , Collections.emptyList()
                , voucherId.toString()
                , userId.toString());
        // 2、判断结果不为 0，没有秒杀资格
        int r = result.intValue();
        if (r != 0) {
            log.error(r == 1 ? "没有库存：{}" : "不能重复下单，{}，{}", voucherId, userId);
            return;
        }
        // 3、判断为0，有资格参与秒杀，设置订单信息。
        Long orderId = redisIdWorker.nextId("order");
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setId(orderId);
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setUserId(userId);
        // 4、将订单信息放进队列，等待异步线程处理。
        orderTasks.offer(voucherOrder);
    }*/


/*    @Override
    public Result seckillVoucher(Long voucherId) {
        // 1、查找优惠卷信息
        SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
        // 2、判断是否在秒杀时间范围，LocalDateTime 是真好用，before，after
        LocalDateTime now = LocalDateTime.now();
        if (now.isBefore(seckillVoucher.getBeginTime()) || now.isAfter(seckillVoucher.getEndTime())) {
            return Result.fail("不在秒杀时间内");
        }
        // 3、判断库存
        if (seckillVoucher.getStock() < 1) {
            return Result.fail("库存不足");
        }
        // 4、一人一单，单机锁，synchronized 悲观锁
//       synchronized (UserHolder.getUser().getId().toString().intern()) {
//            // 代理对象（事务需要），否则直接类自调用，事务注解没有经过接口代理，是不会起作用的
//            IVoucherOrderService voucherOrderService = (IVoucherOrderService) AopContext.currentProxy();
//            return voucherOrderService.createVoucherOrder(voucherId);
//        }
        // 5、集群版，通过 setnx 实现
        Long userId = UserHolder.getUser().getId();
        // 5.1、创建锁对象
//        SimpleRedisLock simpleRedisLock = new SimpleRedisLock("order:" + userId, redisTemplate);
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        // 5.2、获取锁
//        boolean isLock = simpleRedisLock.tryLock(1200L);
        // 参数依次为 waitTime（等待获取锁的时间，可在该时间内重复获取锁）,leaseTime（锁的时间）,TimeUnit（单位）
        // 不写参数则默认只获取一次锁，30s
        boolean isLock = lock.tryLock();
        // 5.3、判断是否拿到锁
        if (!isLock) {
            return Result.fail("不允许重复下单");
        }
        try {
            IVoucherOrderService voucherOrderService = (IVoucherOrderService) AopContext.currentProxy();
            return voucherOrderService.createVoucherOrder(voucherId);
        } finally {
            // 即便 try 的代码有 return，finally 还是会执行
//            simpleRedisLock.unLock();
            lock.unlock();
        }

    }*/

    /**
     * 创建订单，
     *
     * @param voucherId
     * @return
     */
/*    @Transactional
    @Override
    public Result createVoucherOrder(Long voucherId) {
        // 4、一人一单
        Long userId = UserHolder.getUser().getId();
        Integer count = this.query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        if (count > 0) {
            return Result.fail("用户已经购买过一次了");
        }

        // 5、扣减库存（应该是先锁库存，用户支付之后，扣减库存，时间内未支付，库存解锁）
        // 两个 update() 方法是不同的，一个返回值是 UpdateChainWrapper，可以继续执行一些操作。
        // 另一个是 ChainUpdate 的接口默认方法，默认实现了，对 updateChain 的操作进行执行。
        seckillVoucherService.update()
                .setSql("stock = stock - 1")
                .eq("voucher_id", voucherId)
                .gt("stock", 0)
                .update();   // update table set stock = stock - 1 where voucher_id = ? and stock > 0
        // 6、创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        // 6.1、订单id，使用 redis 全局唯一id
        Long orderId = redisIdWorker.nextId(ICR_ORDER_ID_PREFIX);
        voucherOrder.setId(orderId);
        // 6.2、用户id
        voucherOrder.setUserId(userId);
        // 6.3、代金卷id
        voucherOrder.setVoucherId(voucherId);
        voucherOrderService.save(voucherOrder);
        // 7、返回订单id
        return Result.ok(orderId);
    }*/
}
